package com.gct.es;

/**
 * <a href="https://www.elastic.co">ElasticSearch Official web site</a>
 */
import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import org.elasticsearch.action.bulk.BackoffPolicy;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.get.MultiGetItemResponse;
import org.elasticsearch.action.get.MultiGetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.MultiSearchRequestBuilder;
import org.elasticsearch.action.search.MultiSearchResponse;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.MatchQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.sort.SortOrder;
import org.elasticsearch.search.sort.SortParseElement;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.gct.domain.BasePojo;

public class ESTransportClient {
    protected static final Logger logger = LoggerFactory.getLogger(ESTransportClient.class);
    private static TransportClient client;
    private static final ObjectMapper mapper = new ObjectMapper();
    private static final Properties prop = new Properties();
    private final SingleDocument sd = this.new SingleDocument();
    private final MultiDocument md = this.new MultiDocument();

    static{
        try {
            prop.load(ESTransportClient.class.getResourceAsStream("esserverconfig.properties"));
            client = TransportClient.builder().build()
                    .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(prop.getProperty("host")), Integer.valueOf(prop.getProperty("port"))));
        } catch (UnknownHostException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static enum QueryBuilderType {
        TERMQUERY,
        MATCHQUERY
    }

    private class SingleDocument {
        /**
         * Retrieve document from ES node
         *
         * @param index - index name in node
         * @param type - type name in node
         * @param id - document id
         * 
         * @return  document or null
         */
        Map<String, Object> getDocument(String index, String type, String id) {
            GetResponse response = client.prepareGet(index, type, id).setOperationThreaded(false).get();
            return response.getSource();
        }

        void indexDocument(BasePojo bpObj) {
            try {
                byte[] json = mapper.writeValueAsBytes(bpObj);
                client.prepareIndex(getIndex(), bpObj.getClass().getSimpleName().toLowerCase(), String.valueOf(bpObj.getId()))
                    .setSource(json)
                    .get();
            } catch (JsonProcessingException e) {
                e.printStackTrace();
            }
        }

        void indexDocument(Map<String, String> json, String type, String id) {
            client.prepareIndex(getIndex(), type, id).setSource(json).get();
        }

       void deleteDocument(String type, String id) {
            client.prepareDelete(getIndex(), type, id).get();
        }
    }

    public static enum BoolMatchType {
        MUST,
        MUSTNOT,
        FILTER,
        SHOULD
    }

    private class MultiDocument {
        private BulkProcessor bulkProcessor;
        private int bulkAction = 10000;
        private long bulkFlushInterval = 2; /*in seconds*/
        private long bulkDelayInterval = 1000; /*in milliseconds*/
        private int bulkBackoffRetry = 3; /*retry times when failure happens*/
        private long scrollAliveSecond = 60000;
        private int scrollSize = 2;

        List<Map<String, Object>> getDocuments(String index, String type, List<String> id) {
            MultiGetResponse multiGetItemResponses = client.prepareMultiGet().add(index, type, id).get();
            List<Map<String, Object>> rtObj = new LinkedList<Map<String, Object>>();
            for (MultiGetItemResponse itemResponse : multiGetItemResponses) {
                GetResponse response = itemResponse.getResponse();
                rtObj.add(response.getSource());
            }
            return rtObj;
        }

       void indexDocuments(List<? extends BasePojo> bpObjs) {
            try {
                for (BasePojo b : bpObjs) {
                    byte[] json = mapper.writeValueAsBytes(b);
                    bulkProcessor.add(new IndexRequest(getIndex(), b.getClass().getSimpleName().toLowerCase(), String.valueOf(b.getId())).source(json));
                }
                bulkProcessor.flush();
            } catch (JsonProcessingException e) {
                e.printStackTrace();
            }
        }

        void closeBulk() {
            if(null != bulkProcessor)
                bulkProcessor.close();
        }

        QueryBuilder getQuery(String field, Object searchingText, QueryBuilderType queryBuilderType) {
            if(queryBuilderType.equals(QueryBuilderType.TERMQUERY))
                return QueryBuilders.termQuery(field,searchingText);
            if(queryBuilderType.equals(QueryBuilderType.MATCHQUERY))
                if(null != prop.get("matchOperator") && !"".equals(prop.get("matchOperator")))
                    if("AND".equals(prop.get("matchOperator")))
                        return QueryBuilders.matchQuery(field, searchingText).operator(MatchQueryBuilder.Operator.AND);
                    else
                        return QueryBuilders.matchQuery(field, searchingText);
                else if(null != prop.get("minimunMatch") && !"".equals(prop.get("minimunMatch")))
                    return QueryBuilders.matchQuery(field, searchingText).minimumShouldMatch(prop.get("minimunMatch").toString());
                else
                    return QueryBuilders.matchQuery(field, searchingText);

            // default query
            return QueryBuilders.termQuery(field,searchingText);
        }

        /**
         * default modifier, accessible by classes which are in the same package.
         */
        List<Map<String, Object>> scrollMatchDocuments(String field, Object searchingText, String scrollId, QueryBuilderType queryBuilderType, String type) {
            LinkedList<Map<String, Object>> rtObjs = new LinkedList<Map<String, Object>>();
            SearchResponse scrollResp;
            if (null == scrollId) {
                QueryBuilder qb = this.getQuery(field, searchingText, queryBuilderType);
                scrollResp = client.prepareSearch(ESTransportClient.this.getIndex())
                        .setTypes(type)
                        .addSort(SortParseElement.SCORE_FIELD_NAME, SortOrder.ASC)
                        .setScroll(new TimeValue(scrollAliveSecond))
                        .setQuery(qb).setSize(scrollSize).execute().actionGet();
                for (SearchHit hit : scrollResp.getHits().getHits()) {
                    rtObjs.add(hit.getSource());
                }

                Map<String,Object> sId = new HashMap<String,Object>(1);
                sId.put("_scrollId", scrollResp.getScrollId());
                rtObjs.addFirst(sId);
            } else {
                scrollResp = client.prepareSearchScroll(scrollId).setScroll(new TimeValue(scrollAliveSecond)).execute().actionGet();
                for (SearchHit hit : scrollResp.getHits().getHits()) {
                    rtObjs.add(hit.getSource());
                }
            }
            return rtObjs;
        }

        /**
         * with private modifier, still can be used within the outer class with inner class instance, but for other classes out of the outer class.
         */
        private void addScrollId(LinkedList<Map<String, Map<String,Object>>> rtObjs, Object _scrollId) {
            Map<String,Map<String,Object>> sId = new HashMap<String,Map<String,Object>>(1);
            Map<String,Object> mId = new HashMap<String,Object>();
            mId.put("_scrollId", _scrollId);
            sId.put("_scrollId", mId);
            rtObjs.addFirst(sId);
        }

        private void addDocsWithScroll(LinkedList<Map<String, Map<String,Object>>> rtObjs, String scrollId) {
            SearchResponse scr = client.prepareSearchScroll(scrollId).setScroll(new TimeValue(scrollAliveSecond)).execute().actionGet();
            for (SearchHit hit : scr.getHits().getHits()) {
                Map<String, Map<String, Object>> doc = new HashMap<String, Map<String, Object>>(1);
                doc.put(hit.getType() +"_"+ hit.getId(), hit.getSource());
                rtObjs.add(doc);
            }
        }

        List<Map<String,Map<String,Object>>> scrollMultiMatchDocuments(Map<String,Object> criteria, List<String> scrollId, QueryBuilderType queryBuilderType, String... types) {
            LinkedList<Map<String, Map<String,Object>>> rtObjs = new LinkedList<Map<String, Map<String,Object>>>();
            MultiSearchResponse scrollResp;
            if (null == scrollId) {
                MultiSearchRequestBuilder msrb = client.prepareMultiSearch();
                Iterator<String> itr = criteria.keySet().iterator();
                while (itr.hasNext()) {
                    String field = itr.next();
                    QueryBuilder qb = this.getQuery(field, criteria.get(field), queryBuilderType);
                    SearchRequestBuilder srb = client.prepareSearch(ESTransportClient.this.getIndex())
                            .setTypes(types)
                            .setScroll(new TimeValue(scrollAliveSecond))
                            .setQuery(qb).setSize(scrollSize);
                    msrb.add(srb);
                }
                scrollResp = msrb.execute().actionGet();
                List<String> _scrollId = new LinkedList<String>();
                for (MultiSearchResponse.Item item : scrollResp.getResponses()) {
                    for (SearchHit hit : item.getResponse().getHits().getHits()) {
                        Map<String,Map<String,Object>> doc = new HashMap<String,Map<String,Object>>(1);
                        doc.put(hit.getType()+hit.getId(), hit.getSource());
                        rtObjs.add(doc);
                        _scrollId.add(item.getResponse().getScrollId());
                    }
                    logger.info(item.getResponse().getScrollId());
                }

                addScrollId(rtObjs,_scrollId);
            } else {
                for (String id : scrollId) {
                    addDocsWithScroll(rtObjs,id);
                }
            }
            return rtObjs;
        }

        List<Map<String,Map<String,Object>>> scrollBoolMatchDocuments(Map<BoolMatchType,Map<String,Object>> criteria, String scrollId, QueryBuilderType queryBuilderType, String... types) {
            LinkedList<Map<String, Map<String,Object>>> rtObjs = new LinkedList<Map<String, Map<String,Object>>>();
            if (null == scrollId) {
                BoolQueryBuilder bqb = QueryBuilders.boolQuery();
                Iterator<BoolMatchType> itr = criteria.keySet().iterator();
                while (itr.hasNext()) {
                    BoolMatchType boolType = itr.next();
                    Map<String,Object> innerMap = criteria.get(boolType);
                    Iterator<String> innerKey = innerMap.keySet().iterator();
                    while(innerKey.hasNext()) {
                        String field = innerKey.next();
                        QueryBuilder qb = this.getQuery(field, innerMap.get(field), queryBuilderType);
                        switch (boolType) {
                        case SHOULD:
                            bqb.should(qb);
                            break;
                        case MUST:
                            bqb.must(qb);
                            break;
                        case MUSTNOT:
                            bqb.mustNot(qb);
                            break;
                        case FILTER:
                            bqb.filter(qb);
                            break;
                        default:
                            throw new RuntimeException("Illegal Matching Type assigned. Valid types are [should,must,mustnot,filter]");
                        }
                    }
                }
                SearchResponse scrollResp = client.prepareSearch(ESTransportClient.this.getIndex())
                        .setTypes(types)
                        .setScroll(new TimeValue(scrollAliveSecond))
                        .setQuery(bqb).setSize(scrollSize).execute().actionGet();
                String _scrollId = scrollResp.getScrollId();
                for (SearchHit hit : scrollResp.getHits().getHits()) {
                    Map<String, Map<String, Object>> doc = new HashMap<String, Map<String, Object>>(1);
                    doc.put(hit.getType() +"_"+ hit.getId(), hit.getSource());
                    rtObjs.add(doc);
                }

                addScrollId(rtObjs,_scrollId);
            } else {
                addDocsWithScroll(rtObjs,scrollId);
            }
            return rtObjs;
        }

        {
            bulkProcessor = BulkProcessor.builder(client, new BulkProcessor.Listener() {
                @Override
                public void beforeBulk(long executionId, BulkRequest request) {
                    logger.info("before bulk processing...");
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }

                @Override
                public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
                    logger.info("bulk processed...");
                }

                @Override
                public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
                    failure.printStackTrace();
                }
            }).setBulkActions(bulkAction).setBulkSize(new ByteSizeValue(1, ByteSizeUnit.GB)).setFlushInterval(TimeValue.timeValueSeconds(bulkFlushInterval))
                    .setConcurrentRequests(0)
                    .setBackoffPolicy(BackoffPolicy.exponentialBackoff(TimeValue.timeValueMillis(bulkDelayInterval), bulkBackoffRetry)).build();
        }
    }

    public ESTransportClient indexDocument(BasePojo bpObj) {
        sd.indexDocument(bpObj);
        return this;
    }

    public ESTransportClient indexDocument(Map<String, String> json, String type, String id) {
        sd.indexDocument(json, type, id);
        return this;
    }

    public Map<String, Object> getDocument(String type, String id) {
        return sd.getDocument(getIndex(), type, id);
    }

    public ESTransportClient deleteDocument(String type, String id) {
        sd.deleteDocument(type, id);
        return this;
    }

    public List<Map<String, Object>> getDocuments(String type, List<String> id) {
        return md.getDocuments(getIndex(), type, id);
    }

    public ESTransportClient indexDocuments(List<? extends BasePojo> bpObjs) {
        md.indexDocuments(bpObjs);
        return this;
    }

    /**
     * search documents via scroll with single field, one searching text in single type of documents
     *
     * @param field - filed used to search
     * @param searchingText - text used to matching
     * @param scrollId - scrollId returned after the first scroll searching; should be null at the first calling
     * @param types    - document types used to search against
     * @param queryBuilderType - which QueryBuilder to use
     * 
     * @return  a linkedList of documents; the first element contains scrollId after the first calling if matching.
     */
    public List<Map<String, Object>> scrollMatchDocuments(String field, Object searchingText, String scrollId, QueryBuilderType queryBuilderType, String type) {
        return md.scrollMatchDocuments(field, searchingText, scrollId,queryBuilderType, type);
    }

    public void closeBulk() {
        md.closeBulk();
    }

    private String getIndex() {
        return prop.getProperty("index").intern();
    }

    /**
     * search documents via scroll with multiple fields(or single field) in multiple types of documents or single type of documents,
     * each field has a searching text;
     *
     * @param criteria - a map with fields and searching text
     * @param scrollId - a list of scrollIds returned after the first scroll searching; should be null at the first calling
     * @param types    - document types used to search against
     * @param queryBuilderType - which QueryBuilder to use
     * 
     * @return  a linkedList of documents, key is document type; the first element contains scrollId after the first calling if matching, key for that is '_scrollId'.
     */
    public List<Map<String,Map<String,Object>>> scrollMultiMatchDocuments(Map<String,Object> criteria, List<String> scrollId, QueryBuilderType queryBuilderType, String... types) {
        return md.scrollMultiMatchDocuments(criteria, scrollId, queryBuilderType, types);
    }

    /**
     * search documents via bool query with scroll. 
     *
     * @param criteria - a map with fields and searching text grouped by matching type. Valid matching types are BoolMatchType.MUST, BoolMatchType.MUSTNOT,
     *                      BoolMatchType.FILTER, BoolMatchType.SHOULD.
     * @param scrollId - scrollId returned after the first scroll searching; should be null at the first calling.
     * @param types    - document types used to search against.
     * @param queryBuilderType - which QueryBuilder to use, like term query or match query.
     * 
     * @return  a linkedList of documents, key is document type; the first element contains scrollId after the first calling if matching, key for that is '_scrollId'.
     */
    public List<Map<String,Map<String,Object>>> scrollBoolMatchDocuments(Map<BoolMatchType,Map<String,Object>> criteria, String scrollId, QueryBuilderType queryBuilderType, String... types) {
        return md.scrollBoolMatchDocuments(criteria, scrollId, queryBuilderType, types);
    }
}
